/*
 * Copyright 2016-present the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.github.pnoker.common.mqtt.handler;

import cn.hutool.core.text.CharSequenceUtil;
import io.github.pnoker.common.mqtt.entity.MessageHeader;
import io.github.pnoker.common.mqtt.entity.MqttMessage;
import io.github.pnoker.common.mqtt.job.MqttScheduleJob;
import io.github.pnoker.common.mqtt.service.MqttReceiveService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.MessageHandler;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author pnoker
 * @since 2022.1.0
 */
@Slf4j
@Component
//该控制层用于MQTT驱动进行接收数据，数据处理正常应该在dc3-driver-mqtt的impl实现类里面实现
public class MqttReceiveHandler {

    @Value("${driver.mqtt.batch.speed}")
    private Integer batchSpeed;

    @Resource
    private MqttReceiveService mqttReceiveService;
    @Resource
    private ThreadPoolExecutor threadPoolExecutor;

    /**
     * 此处用于接收 MQTT 发送过来的数据，订阅的主题为 application.yml 中 mqtt.receive-topics 配置的 Topic 列表
     * +（加号）：可以（只能）匹配一个单词
     * #（井号）：可以匹配多个单词（或者零个）
     *
     * @return MessageHandler
     */
//    @Bean 这个注解表示将下面的方法返回的对象注册为一个bean，可以被其他组件引用。
    @Bean
//    这个注解表示将消息处理器绑定到名为"mqttInboundChannel"的通道上，表示接收该通道上的消息并进行处理。
    @ServiceActivator(inputChannel = "mqttInboundChannel")
//    mqttInboundReceive 这是一个方法，返回类型为MessageHandler。因为上面使用了@Bean注解，所以此方法将被调用，创建并返回一个消息处理器。
    public MessageHandler mqttInboundReceive() {
        return message -> {                 //这是一个Lambda表达式，定义了一个匿名的消息处理函数。接收到的消息会经过这个处理函数进行处理。
            try {
                MessageHeader messageHeader = new MessageHeader(message.getHeaders());   //创建一个MessageHeader对象，从接收到的消息中获取相关的消息头信息。
                String payload = message.getPayload().toString(); //获取接收到的消息体，并将其转换为字符串形式。
                if (CharSequenceUtil.isEmpty(payload)) {   //判断接收到的消息体是否为空，如果为空，则记录错误日志并返回。
                    log.error("Invalid mqtt inbound, From: {}, Qos: {}, Payload: null", messageHeader.getMqttReceivedTopic(), messageHeader.getMqttReceivedQos());
                    return;
                }
                MqttScheduleJob.messageCount.getAndIncrement();    //增加一个计数器，用于统计接收到的消息数量。
                MqttMessage mqttMessage = new MqttMessage(messageHeader, payload);  //创建一个MqttMessage对象，将消息头和消息体封装起来。
                log.debug("Mqtt inbound, From: {}, Qos: {}, Payload: {}", messageHeader.getMqttReceivedTopic(), messageHeader.getMqttReceivedQos(), payload);

                // 根据数据传输速度判断是否要批量处理数据。如果传输速度低于阈值，则使用线程池异步执行单个MQTT消息的处理逻辑。
                // Judge whether to process data in batch according to the data transmission speed
                if (MqttScheduleJob.messageSpeed.get() < batchSpeed) {
                    threadPoolExecutor.execute(() ->
                            // Receive single mqtt message
                            //使用线程池异步执行mqttReceiveService.receiveValue(mqttMessage)方法，实际处理单个MQTT消息的业务逻辑。
                            mqttReceiveService.receiveValue(mqttMessage)
                    );
                } else {  //如果传输速度高于或等于阈值，将消息添加到一个消息队列中，稍后进行批量处理。
                    // Save point value to schedule
                    MqttScheduleJob.messageLock.writeLock().lock();  //获取写锁，用于对消息队列的操作。
                    MqttScheduleJob.addMqttMessages(mqttMessage);  //将消息添加到消息队列中。
                    MqttScheduleJob.messageLock.writeLock().unlock();  // 释放写锁。
                }
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        };
    }
}
